// const fs = require('fs')
// const rs = fs.createReadStream('./06.txt')
// const ws = fs.createWriteStream('./061.txt')
// rs.pipe(ws)

// const {Readable} = require('stream')
// class ArrayReadable extends Readable{
//     constructor(source){
//         super()
//         this.source = source
//     }
//     _read(){
//         // 底层数据被读取完，传递 null，内部实现就会知道数据读取完毕
//         const data = this.source.shift() || null
//         this.push(data)
//     }
// }
// const arrayReadable = new ArrayReadable(['a','b','c','d','e','f'])
// // data、readable 消费数据，readable 需要主动调用 read 方法。
// arrayReadable.on('data',data=>{
//     console.log(data.toString())
// })
// arrayReadable.on('readable',()=>{
//     let data = null
//     while((data = arrayReadable.read()) !== null){
//         console.log(data.toString())
//     }
// })
// // 停止时调用
// arrayReadable.on('end',()=>{
//    console.log('end')
// })

// const {Writable} = require('stream')
// class StdoutWriteable extends Writable{
//     constructor(){
//         super()
//     }
//     // chunk：数据、en：编码、done：回调
//     _write(chunk,en,done){
//         process.stdout.write(chunk.toString() + '<---')
//         // 同步代码执行完成才回调
//         process.nextTick(done)
//     }
// }
// const stdoutWriteable = new StdoutWriteable()
// stdoutWriteable.write('哈哈哈','utf-8',()=>{
//     console.log('写入成功')
// })
// // 哈哈哈---写入成功

// const {Duplex} = require('stream')

// class MyDuplex extends Duplex{
//     constructor(source){
//         super()
//         this.source = source
//     }
//     _read(){
//         const data = this.source.shift() || null
//         this.push(data)
//     }
//     _write(chunk,en,done){
//         process.stdout.write(chunk.toString() + '<---')
//         process.nextTick(done)
//     }
// }
// const myDuplex = new MyDuplex(['a','b','c','d','e','f'])
// myDuplex.on('data',data=>{
//     console.log(data.toString())
// })
// myDuplex.write('哈哈哈')

// const {Transform} = require('stream')

// class MyTransform extends Transform{
//     constructor(){
//         super()
//     }
//     _transform(chunk,en,cb){
//         this.push(chunk.toString().toUpperCase())
//         cb(null)
//     }
// }
// let transform = new MyTransform()
// transform.write('aaa')
// transform.on('data',(chunk)=>{
//     console.log(chunk.toString())
// })




// const fs = require('fs')
// const rs = fs.createReadStream('./06.txt',{
//     flags:'r', 
//     encoding:'utf-8', 
//     fd:null, // 文件描述符，默认从 3 开始
//     mode: 438, // 权限
//     autoClose:true, // 是否自动关闭
//     start:0, // 开始位置
//     // end:3, // 结束位置
//     highWaterMark: 4 // 一次读取 2 个字节
// })
// // rs.on('data',(chunk)=>{
// //     console.log(chunk.toString())
// //     rs.pause() // 暂停
// //     setTimeout(()=>{
// //         rs.resume() // 重新开始
// //     },1000)
// // })

// // rs.on('readable',()=>{
// //     // read：可以设置从缓冲区读取字节，缓冲区为空时再读取
// //     let data
// //     while(( data = rs.read(1)) !== null){
// //         console.log(data.toString(),rs._readableState.length)
// //     }
// // })

// // 默认情况下会触发
// rs.on('open',(fd)=>{
//     console.log('文件打开',fd)
// })
// // 默认可读流是暂停模式，不会触发，有 data 事件后才会触发
// rs.on('close',()=>{
//     console.log('文件关闭')
// })
// rs.on('data',()=>{
//     console.log('data')
// })
// // 数据消费完成之后触发，close 之前调用
// rs.on('end',()=>{
//     console.log('end')
// })
// rs.on('error',()=>{
//     console.log('error')
// })

// const fs = require('fs')
// const ws = fs.createWriteStream('./061.txt',{
//     flags:'w',
//     mode:438,
//     fd:null,
//     encoding:'utf-8',
//     start: 0,
//     highWaterMark:3
// })

// ws.write('哈哈哈',()=>{
//     console.log('写完了1')
// })

// // 异步操作会串行
// // ws.write(Buffer.from('qqq'),()=>{
// //     console.log('写完了2')
// // })

// ws.on('open',(fd)=>{
//     console.log('open',fd)
// })
// // end 之后执行
// ws.on('close',()=>{
//     console.log('close')
// })
// // 数据写入完成，end 之后不能写入
// ws.end()

// ws.on('error',()=>{
//     console.log('error')
// })

// const fs = require('fs')
// const ws = fs.createWriteStream('./061.txt',{
//     highWaterMark: 3
// })
// // flag 为 flase 不意味不能写入了，仅仅用于提示消费速度跟不上生产速度了
// // 可读流的模式可以修改为暂停，等消费者消费完成之后通过 drain 通知

// // let flag = ws.write('a')
// // console.log(flag)
// // flag = ws.write('b')
// // console.log(flag)
// // flag = ws.write('c')
// // console.log(flag)
// let source = '哈哈哈哈哈'.split('')

// let num = 0
// let  flag = true
// function executeWrite(){
//     while(num < source.length && flag){
//         flag = ws.write(source[num])
//         num ++
//     }
// }
// executeWrite()

// // drain 用于分批写入（防止内存溢出）、限流写入速度
// // 写入的不小于 highWaterMark 时触发，可以使用 pipe 实现
// ws.on('drain',()=>{
//     console.log('drain')
//     flag = true
//     executeWrite()
// })



// const fs = require('fs')
// const rs = fs.createReadStream('./06.txt',{
//     highWaterMark: 3
// })
// const ws = fs.createWriteStream('./061.txt',{
//     highWaterMark: 3
// })
// let flag = true
// rs.on('data',(chunk)=>{
//     flag = ws.write(chunk,()=>{
//         console.log('OK')
//     })
//     console.log(flag)
//     if(!flag){
//         rs.pause()
//     }
// })
// // drain 用于分批写入（防止内存溢出）、限流写入速度
// // 缓冲区可写时触发
// ws.on('drain',()=>{
//     rs.resume()
// })

const fs = require('fs')
const {EventEmitter} = require('events')

class FileReadStream extends EventEmitter{
    constructor(path,opts={}){
        super()
        this.path = path
        this.flags = opts.flags || 'r'
        this.mode = opts.mode || 438
        this.autoClose =  opts.autoClose || true
        this.start = opts.start || 0
        this.end = opts.end 
        this.highWaterMark = opts.highWaterMark || 64*1024 // 可读流默认是 64k
        this.readOffset = 0

        this.flowing = false 
        this.open()
        // newListener：监听事件时触发
        this.on('newListener',(type)=>{
            if(type === 'data'){
                this.flowing = true
                this.read()
            }
        })
    }
    open(){
        fs.open(this.path,this.flags,this.mode,(err,fd)=>{
            if(err){
                this.emit('error',err)
                return
            }
            this.fd = fd
            this.emit('open',fd)
        })
    }
    read(){
        // fs.open 是异步的，所以需要订阅才能得到 fd
        if(typeof this.fd !== 'number'){
            return this.once('open',this.read)
        }
        // 读取字节数
        let howMuchToRead  = this.end ? Math.min(this.end - this.readOffset, this.highWaterMark) : this.highWaterMark
        let buf = Buffer.alloc(this.highWaterMark)
        fs.read(this.fd,buf,0,howMuchToRead,this.readOffset,(err,bytes)=>{
            // 还有数据，bytes：读取字节数
            if(bytes){
                this.readOffset += bytes
                this.emit('data',buf.slice(0,bytes)) // 防止有多余的空白 buffer 位
                this.flowing && this.read()
                return
            }
            this.emit('end')
            this.close()
        })
    }
    close(){
        fs.close(this.fd,()=>{
           this.emit('close')
        })
    }
    resume(){
        this.flowing=true
        this.read()
    }
    pause(){
        this.flowing=false
    }
}
// const  fileReadStream = new FileReadStream('061.txt',{
//     highWaterMark: 3,
//     end: 3
// })
// fileReadStream.on('open',(fd)=>{
//     console.log('open',fd)
// })
// fileReadStream.on('error',(error)=>{
//     console.log(error)
// })
// fileReadStream.on('data',(chunk)=>{
//     console.log(chunk.toString())
// })
// fileReadStream.on('end',()=>{
//     console.log('end')
// })
// fileReadStream.on('close',()=>{
//     console.log('close')
// })


class Node {
    constructor(element,next){
        this.element = element
        this.next = next
    }
}
class LinkList {
    constructor(){
        this.head = null
        this.size = 0
    }
    add(index,element){
        if(arguments.length === 1){
            element = index
            index = this.size
        }
        if(index < 0 || index > this.size){
            throw new Error('cross the border')
        }
        if(index === 0){
            let head = this.head
            this.head = new Node(element,head)
        }else{
            let preNode = this._getNode(index - 1)
            preNode.next = new Node(element,preNode.next)
        }
        this.size ++
    }
    _getNode(index){
        let node = this.head
        for (let i = 0; i < index; i++) {
            node = node.next
        }
        return node
    }
    remove(index){
        let rmNode = null
        if(index === 0){
            rmNode = this.head
            if(!rmNode){
                return undefined
            }
            this.head = rmNode.next
        }else{
            const preNode = this._getNode(index - 1)
            rmNode = preNode.next
            preNode.next = rmNode.next
        }
        this.size --
        return rmNode
    }
    set(index,element){
        const node = this._getNode(index)
        node.element = element
    }
    get(index){
        return this._getNode(index)
    }
    clear(){
        this.head = null
        this.size = 0
    }
}
class Queue{
    constructor(){
        this.linkList = new LinkList()
    }
    enQueue(data){
        this.linkList.add(data)
    }
    deQueue(){
        return this.linkList.remove(0)
    }
}
// const list = new LinkList()
// list.add('a')
// list.add('b')
// list.add(1,'c')
// list.set(1,'caaa')
// list.remove(2)
// console.log(list)
// const queue = new Queue()
// queue.enQueue('node1')
// queue.enQueue('node2')
// queue.enQueue('node3')
// queue.deQueue()
// queue.deQueue()
// console.log(queue.deQueue())

// const {EventEmitter} = require('events')
// const fs = require('fs')

class FileWriteStream extends EventEmitter{
    constructor(path,opts={}){
        super()
        this.path = path
        this.flags = opts.flags || 'w'
        this.mode = opts.mode || 438
        this.autoClose =  opts.autoClose || true
        this.start = opts.start || 0
        this.encoding = opts.encoding || 'utf8'
        this.highWaterMark = opts.highWaterMark || 16*1024 // 可读流默认是 16k

        this.writeOffset = this.start // 偏移量
        this.writing = false
        this.writeLen = 0
        this.needDrain = false
        this.cache = new Queue()
        this.open()
    }
    open(){
        fs.open(this.path,this.flags,this.mode,(err,fd)=>{
            if(err){
                this.emit('error',err)
                return
            }
            this.fd = fd
            this.emit('open',fd)
        })
    }
    write(chunk,encoding,cb){
        chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)

        this.writeLen += chunk.length
        let flag = this.writeLen < this.highWaterMark 
        this.needDrain = !flag
        if(this.writing){
            // 正在写入，需要排队
            this.cache.enQueue({chunk,encoding,cb})
        }else{
            this.writing = true
            // 读取没有执行写入
            this._write(chunk,encoding,()=>{
                cb && cb()
                // 清空排队内容
                this._clearBuffer()
            })
        }
        return flag
    }
    _write(chunk,encoding,cb){
        if(typeof this.fd !== 'number'){
            return this.once('open',()=>{this._write(chunk,encoding,cb)})
        }
       fs.write(this.fd,chunk,this.start,chunk.length,this.writeOffset,(err,bytes)=>{
            this.writeOffset += bytes
            this.writeLen -= bytes
            cb && cb()
       })
    }
    _clearBuffer(){
        let data = this.cache.deQueue()
        if(data){
           const {chunk,encoding,cb} = data.element
           this._write(chunk,encoding,()=>{
                cb && cb()
                this._clearBuffer()
            })
        }else{
            if(this.needDrain){
                this.needDrain = false
                this.emit('drain')
                this.writing = false
            }
        }
    }
}

// const ws = new FileWriteStream('061.txt',{
//     highWaterMark:3
// })

// ws.on('open',(fd)=>{
//     console.log(fd)
// })
// ws.write('哈哈哈','utf8',()=>{
//     console.log('1')
// })
// ws.on('drain',()=>{
//     console.log('drain')
// })
// ws.write('ADS','utf8',()=>{
//     console.log('2')
// })


FileReadStream.prototype.pipe = function(ws){
    this.on('data',(chunk)=>{
        let flag = ws.write(chunk,'uft-8',()=>{})
        console.log(chunk.toString())
        if(!flag){
            this.pause()
        }
    })
    ws.on('drain',()=>{
        this.resume()
    })
}

const  fileReadStream = new FileReadStream('06.txt',{
    highWaterMark: 3
})

const fileWriteStream = new FileWriteStream('061.txt',{
    highWaterMark:1
})

fileReadStream.pipe(fileWriteStream)
